package com.zy.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.zy.gmall.realtime.bean.DimJoinFunction;
import com.zy.gmall.realtime.common.GmallConfig;
import com.zy.gmall.realtime.util.DimUtil;
import com.zy.gmall.realtime.util.HbaseUtil;
import com.zy.gmall.realtime.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public abstract class DimMapFunction<T> extends RichMapFunction<T,T> implements DimJoinFunction {
    //获取连接
    private Jedis jedis;
    private Connection conn;
    private String tableName;

    public DimMapFunction(String tableName){
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = RedisUtil.getJedis();
        conn = HbaseUtil.getConnect();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeJedis(jedis);
        HbaseUtil.closeHbase(conn);
    }

    @Override
    public T map(T t) throws Exception {
        //先去redis中拿数据，没有的话
        String rowKey = getRowKey(t);
        JSONObject dimInfo = DimUtil.getDimInfo(jedis, conn, GmallConfig.HBASE_NAMESPACE, tableName, rowKey);
        //将获取的数据放到实体类
        if (dimInfo != null){
            join(t, dimInfo);
        }
        return t;
    }

    //因为在这也无法获取到T的类型，可以抽象
//    public abstract join(T t, JSONObject dimInfo);
//
//    private abstract String getRowKey() {
//        return null;
//    }
}
